package com.xqcx.libra.event;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.xqcx.libra.util.AbstractThreadHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class MessageEvent extends AbstractThreadHandler {

    private BlockingQueue<Map<String, Object>> queue;
    private MQProducer producer;

    private String topic;
    private String tag;
    Random random = new Random();

    public MessageEvent(String topic, String tag, BlockingQueue<Map<String, Object>> queue, MQProducer producer) {
        this.topic = topic;
        this.tag = tag;
        this.producer = producer;
        this.queue = queue;
    }

    @Override
    protected void handle() {
        String shardingKey = "";
        try {
            Map<String, Object> take = queue.take();
            String json = JSON.toJSONString(take);
            if (take.get("primaryKey") != null) {
                shardingKey = take.get("primaryKey") + "";
            } else {
                shardingKey = random.nextInt(100) + "";
            }
            Message msg = new Message(topic, tag, shardingKey, json.getBytes("UTF-8"));

            log.info("mq:{}",json);
            // 发送顺序消息
//            SendResult send = producer.send(msg, new MessageQueueSelector() {
//                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
//                    Long id = (Long) o;
//                    int index = (int) (id % Long.parseLong(list.size() + ""));
//                    return list.get(index);
//                }
//            }, Long.parseLong(shardingKey));
        } catch (Exception e) {
            log.error("MessageEvent.handle topic:{} shardingKey:{} Exception:{}", topic, shardingKey, e);
        }
    }

}
